package datastream.api.operator.one;

import datastream.api.operator.one.source.log.LogSource;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import datastream.api.operator.one.pojo.log.InputModel;
import datastream.api.operator.one.pojo.log.OutputModel;

// 将一条批量上报的页面点击日志转换为多条日志，且只包含页面 1 的点击日志
public class FlatMap {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        DataStreamSource<InputModel> source = env.addSource(new LogSource());

        SingleOutputStreamOperator<OutputModel> flatMap = source.flatMap(new FlatMapFunction<InputModel, OutputModel>() {
            @Override
            public void flatMap(InputModel value, Collector<OutputModel> collector) throws Exception {

                for (Tuple2<String, String> log : value.getBatchLog()) {
                    if ("页面1".equals(log.f0)) {
                        collector.collect(OutputModel.builder()
                                .username(value.getUsername())
                                .log(log)
                                .build());
                    }
                }
            }
        });

        DataStreamSink<OutputModel> sink = flatMap.print();

        env.execute();
    }
}
